package com.wuwangfu.process;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.OutputTag;

import java.time.Duration;

/**
 * @Author: jcshen
 * @Date: 2023-03-08
 * @PackageName: com.wuwangfu.process
 * @ClassName: WindowOutputTag
 * @Description: 测流输出获取窗口迟到数据
 * @Version: 1.0.0
 */
public class WindowOutputTag {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        1000,spark,1
//        2000,flink,1
//        4000,hive,1
//        4800,spark,2
//        4999,flink,2
//        5000,hive,2
//        4888,hadoop,6
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);
        //生成watermark
        SingleOutputStreamOperator<String> dataWatermark = lines.assignTimestampsAndWatermarks(WatermarkStrategy
                .<String>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner((line, timestamp) ->
                        Long.parseLong(line.split(",")[0])));
        //
        SingleOutputStreamOperator<Tuple2<String, Integer>> maped = dataWatermark.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] fields = value.split(",");

                return Tuple2.of(fields[1], Integer.parseInt(fields[2]));
            }
        });
        //分组
        KeyedStream<Tuple2<String, Integer>, String> keyed = maped.keyBy(t -> t.f0);
        //侧输出流
        OutputTag<Tuple2<String, Integer>> lateData = new OutputTag<Tuple2<String, Integer>>("late") {
        };
        //开窗
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> windowed = keyed
                //滚动窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //将延迟数据打上标签
                .sideOutputLateData(lateData);
        //聚合
        SingleOutputStreamOperator<Tuple2<String, Integer>> sumed = windowed.sum(1);
        //主流
        sumed.print("sum：");
        //侧流
        DataStream<Tuple2<String, Integer>> lateStream = sumed.getSideOutput(lateData);
        lateStream.print("late：");

        env.execute();
    }
}
